package com.brid.KafkaProducer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * @Author: 司军强
 * @Date: 14:21 2019/12/19
 * @Description: 自定义kafka生产者拦截器
 */
@Slf4j
public class ProducerInterceptorPrefix implements ProducerInterceptor<String, String> {

    /*todo 定义两个volatile 修饰的变量  ？？？
     */
    private volatile long sendSuccess = 0;
    private volatile long sendFailiure = 0;

    /**
     * 创建一个消息记录，发送
     *
     * @param record
     * @return
     */
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        String modifieValue = "prefix1" + record.value();
        return new ProducerRecord<>(record.topic(), record.partition(), record.timestamp(), record.key(), modifieValue, record.headers());
    }

    /**
     * 发送成功或者失败都会调用这个方法
     *
     * @param metadata
     * @param exception
     */
    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        if (exception == null) {
            sendSuccess++;
        } else {
            sendFailiure++;
        }
    }

    /**
     * 关闭拦截器时调用这个方法
     */
    @Override
    public void close() {
        double successRatio = (double) sendSuccess / (sendFailiure + sendSuccess);
        log.info("【INFO】发送成功率="+String.format("%f",successRatio*100)+"%");
    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
